package com.jingli.pile.kafka;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.jingli.pile.configuration.PushProperties;
import com.jingli.pile.constant.Constants;
import com.jingli.pile.converter.ConnectorConverter;
import com.jingli.pile.converter.EquipmentConverter;
import com.jingli.pile.converter.OrderInfoConverter;
import com.jingli.pile.converter.StationConverter;
import com.jingli.pile.domain.*;
import com.jingli.pile.entity.*;
import com.jingli.pile.service.*;
import com.jingli.pile.util.HmacMd5Signature;
import com.jingli.pile.util.PileSignUtils;
import com.jingli.pile.util.SeqUtils;
import com.jingli.pile.util.ThirdParty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.core.userdetails.UserDetails;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.*;

@Component
public class KafkaMessageConsumer {
    @Autowired
    private PlotService plotService;
    @Autowired
    private SupStationInfoService stationInfoService;
    @Autowired
    private ChargingPileService pileService;
    @Autowired
    private SupEquipmentInfoService equipmentInfoService;
    @Autowired
    private ChargingPortService portService;
    @Autowired
    private SupConnectorInfoService connectorInfoService;
    @Autowired
    private ChargingOrderService chargingOrderService;
    @Autowired
    private SupChargeOrderInfoService orderInfoService;
    @Autowired
    private StationConverter stationConverter;
    @Autowired
    private EquipmentConverter equipmentConverter;
    @Autowired
    private ConnectorConverter connectorConverter;
    @Autowired
    private OrderInfoConverter orderInfoConverter;
    @Autowired
    private PlatformInfoService platformInfoService;
    @Autowired
    private ThirdParty thirdParty;
    @Autowired
    private SupEquipChargeStatusService chargeStatusService;
    @Autowired
    private SupConnectorStatusInfoService connectorStatusInfoService;
    @Autowired
    private PushProperties pushProperties;
    @Autowired
    private PushRecordService pushRecordService;
    private Logger logger = LoggerFactory.getLogger(KafkaMessageConsumer.class);

    //运营商信息
//    @KafkaListener(topics = "${topic.name.operator}")
    public void consumeOperator(String content){
        logger.info("收到代理商信息:{}",content);
    }

    //站点信息
//    @KafkaListener(topics = "${topic.name.station}")
    public void consumeStation(String content){
        logger.info("收到站点信息商信息:{}",content);
        Plot plot = plotService.getById(content);
        SupStationInfo stationInfo = stationConverter.convertPlotToStationInfo(plot);
        logger.info("转换后的站点信息:{}", JSONObject.toJSONString(stationInfo));
        QueryWrapper<SupStationInfo> stationInfoQueryWrapper = new QueryWrapper<>();
        stationInfoQueryWrapper.eq("plot_id",content);
        SupStationInfo supStationInfo = stationInfoService.getOne(stationInfoQueryWrapper);
        if (null==supStationInfo){
            stationInfoService.save(stationInfo);
        }else {
            stationInfoService.updateById(stationInfo);
        }
    }

    //运营商信息
//    @KafkaListener(topics = "${topic.name.equipment}")
    public void consumeEquipment(String content){
        logger.info("收到设备信息商信息:{}",content);
        ChargingPile chargingPile = pileService.getById(content);
        SupEquipmentInfo supEquipmentInfo = equipmentConverter.convertEquipFromPile(chargingPile);
        logger.info("转换后的设备信息:{}", JSONObject.toJSONString(supEquipmentInfo));
        SupEquipmentInfo equipmentInfo = equipmentInfoService.getById(content);
        if (null==equipmentInfo){
            equipmentInfoService.save(supEquipmentInfo);
        }else {
            equipmentInfoService.updateById(supEquipmentInfo);
        }
    }

    //运营商信息
//    @KafkaListener(topics = "${topic.name.connector}")
    public void consumeConnector(String content){
        logger.info("收到枪口信息:{}",content);
        KafkaConnector kafkaConnector = JSONObject.parseObject(content, KafkaConnector.class);
        QueryWrapper<ChargingPort> portQueryWrapper = new QueryWrapper<>();
        portQueryWrapper.eq("deviceid",kafkaConnector.getDeviceId());
        portQueryWrapper.eq("parkid",kafkaConnector.getPileId());
        ChargingPort chargingPort = portService.getOne(portQueryWrapper);
        SupConnectorInfo connectorInfo = connectorConverter.convertConnectorFromPort(chargingPort);
        logger.info("转换后的端口信息:{}", JSONObject.toJSONString(connectorInfo));
        SupConnectorInfo supConnectorInfo = connectorInfoService.getById(content);
        if (null==supConnectorInfo){
            connectorInfoService.save(connectorInfo);
        }else {
            connectorInfoService.updateById(connectorInfo);
        }
    }

    //运营商信息
//    @KafkaListener(topics = "${topic.name.order}")
    public void consumeOrder(String content) throws Exception {
        logger.info("收到订单信息:{}",content);
        QueryWrapper<ChargingOrder> orderQueryWrapper = new QueryWrapper<>();
        orderQueryWrapper.eq("ordernumber",content);
        ChargingOrder chargingOrder = chargingOrderService.getOne(orderQueryWrapper);
        SupChargeOrderInfo orderInfo = orderInfoConverter.convert(chargingOrder);
        logger.info("转换后的订单信息:{}", JSONObject.toJSONString(orderInfo));
        QueryWrapper<SupChargeOrderInfo> orderInfoQueryWrapper = new QueryWrapper<>();
        orderInfoQueryWrapper.likeLeft("order_no",content);
        SupChargeOrderInfo chargeOrderInfo = orderInfoService.getOne(orderInfoQueryWrapper);
        if (null==chargeOrderInfo){
            orderInfoService.save(orderInfo);
        }else {
            orderInfoService.saveOrUpdate(orderInfo);
        }
        PushRecord pushRecord = pushRecordService.getById(content);
        if (null!=pushRecord){
            logger.info("订单:{}已推送,不重复推送,:{}",content,JSONObject.toJSONString(pushRecord));
            return;
        }
        QueryWrapper<PlatformInfo> platformInfoQueryWrapper = new QueryWrapper<>();
        platformInfoQueryWrapper.eq("push_enable","1");
        platformInfoQueryWrapper.eq("env",pushProperties.getEnv());
        List<PlatformInfo> platformInfos = platformInfoService.list(platformInfoQueryWrapper);
        for (PlatformInfo platformInfo :platformInfos) {
            logger.info("原始订单数据:{}",JSONObject.toJSONString(orderInfo));
            String timeStamp = String.valueOf(new Date().getTime());
            String seq = SeqUtils.getUniqueInstance().getSeq();
            String signParams = PileSignUtils.signParams(JSONObject.toJSONString(orderInfo), platformInfo.getDataSecret(), platformInfo.getDataSecretIv());
            String sign = HmacMd5Signature.hmacMd5(platformInfo.getJingliPlatformId()+signParams+ timeStamp + seq,platformInfo.getSignSecret());
            CommonDataResult commonDataResult = new CommonDataResult();
            commonDataResult.setPlatformId(platformInfo.getJingliPlatformId());
            commonDataResult.setSeq(seq);
            commonDataResult.setTimeStamp(timeStamp);
            commonDataResult.setRet(0);
            commonDataResult.setMsg("");
            commonDataResult.setData(signParams);
            commonDataResult.setSig(sign);
            String orderResult = thirdParty.notifyChargeOrderInfo(platformInfo.getPlatformId(), platformInfo.getApiUrl(), JSONObject.toJSONString(commonDataResult));
            logger.info("订单推送结果:{}",JSONObject.toJSONString(orderResult));
            pushRecord = new PushRecord();
            pushRecord.setPushId(content);
            pushRecord.setPushParam(JSONObject.toJSONString(orderInfo));
            pushRecord.setPushResult(orderResult);
            pushRecord.setPushTime(new Date());
            pushRecord.setPushType("order");
            pushRecordService.saveOrUpdate(pushRecord);
        }
    }

    //推送充电站统计信息
    public void pushOperatorStatsInfo(SupStationStatsPushVo supStationStatsPushVo) throws Exception {
        logger.info("准备推送的业务统计数据:{}",JSONObject.toJSONString(supStationStatsPushVo));
        QueryWrapper<PlatformInfo> platformInfoQueryWrapper = new QueryWrapper<>();
        platformInfoQueryWrapper.eq("push_enable","1");
        platformInfoQueryWrapper.eq("env",pushProperties.getEnv());
        List<PlatformInfo> platformInfos = platformInfoService.list(platformInfoQueryWrapper);
        for (PlatformInfo platformInfo :platformInfos) {
            String timeStamp = String.valueOf(new Date().getTime());
            String seq = SeqUtils.getUniqueInstance().getSeq();
            String signParams = PileSignUtils.signParams(JSONObject.toJSONString(supStationStatsPushVo), platformInfo.getDataSecret(), platformInfo.getDataSecretIv());
            String sign = HmacMd5Signature.hmacMd5(platformInfo.getJingliPlatformId()+signParams+timeStamp+seq,platformInfo.getSignSecret());
            CommonDataResult commonDataResult = new CommonDataResult();
            commonDataResult.setPlatformId(platformInfo.getJingliPlatformId());
            commonDataResult.setSeq(seq);
            commonDataResult.setTimeStamp(timeStamp);
            commonDataResult.setRet(0);
            commonDataResult.setMsg("");
            commonDataResult.setData(signParams);
            commonDataResult.setSig(sign);
            logger.info("加密后的数据:{}",JSONObject.toJSONString(commonDataResult));
            String notifyOperationStats = thirdParty.notifyOperationStats(platformInfo.getPlatformId(), platformInfo.getApiUrl(), JSONObject.toJSONString(commonDataResult));
            logger.info("站点统计推送结果:{}",JSONObject.toJSONString(notifyOperationStats));
        }
    }
    //推送充电状态
//    @KafkaListener(topics = "${topic.name.chargeStatus}")
    public void pushChargeStatus(String content) throws Exception {
        KafkaChargeStatus kafkaChargeStatus = JSONObject.parseObject(content, KafkaChargeStatus.class);
        QueryWrapper<ChargingOrder> orderQueryWrapper1 = new QueryWrapper<>();
        orderQueryWrapper1.eq("ordernumber",kafkaChargeStatus.getOrderNo());
        ChargingOrder chargingOrder1 = chargingOrderService.getOne(orderQueryWrapper1);
        SupChargeOrderInfo orderInfo1 = orderInfoConverter.convert(chargingOrder1);
        logger.info("转换后的订单信息:{}", JSONObject.toJSONString(orderInfo1));
        boolean b = orderInfoService.saveOrUpdate(orderInfo1);
        QueryWrapper<SupChargeOrderInfo> orderInfoQueryWrapper = new QueryWrapper<>();
        orderInfoQueryWrapper.likeLeft("order_no",content);
        SupChargeOrderInfo chargeOrderInfo = orderInfoService.getOne(orderInfoQueryWrapper);
        SupEquipChargeStatus chargeStatus = new SupEquipChargeStatus();
        QueryWrapper<SupChargeOrderInfo> orderQueryWrapper = new QueryWrapper<>();
        orderQueryWrapper.likeLeft("order_no",kafkaChargeStatus.getOrderNo());
        SupChargeOrderInfo chargingOrder = orderInfoService.getOne(orderQueryWrapper);
        chargeStatus.setOrderNo(chargingOrder.getOrderNo());
        chargeStatus.setBatteryPackId("");
        chargeStatus.setOrderStatus(2);
        chargeStatus.setOperatorId(chargingOrder.getOperatorId());
        chargeStatus.setEquipmentOwnerId(chargingOrder.getEquipmentOwnerId());
        chargeStatus.setStationId(chargingOrder.getStationId());
        chargeStatus.setEquipmentId(chargingOrder.getEquipmentId());
        chargeStatus.setConnectorId(chargingOrder.getConnectorId());
        chargeStatus.setEquipmentClassification(1);
        chargeStatus.setPushTimeStamp(DateUtil.formatDateTime(new Date()));
        chargeStatus.setConnectorStatus(3);
        chargeStatus.setCurrentA(kafkaChargeStatus.getCurrent());
        chargeStatus.setVoltageA(kafkaChargeStatus.getVoltage());
        chargeStatus.setSoc(kafkaChargeStatus.getSoc());
        chargeStatus.setStartTime(DateUtil.formatDateTime(DateUtil.offsetMinute(new Date(),-1)));
        chargeStatus.setEndTime(DateUtil.formatDateTime(new Date()));
        chargeStatus.setTotalPower(kafkaChargeStatus.getTotalPower());
        chargeStatus.setCurrentMeter(kafkaChargeStatus.getCurrentMeter());
        chargeStatusService.save(chargeStatus);
        logger.info("准备推送充电状态信息:{}",JSONObject.toJSONString(chargeStatus));
        QueryWrapper<PlatformInfo> platformInfoQueryWrapper = new QueryWrapper<>();
        platformInfoQueryWrapper.eq("push_enable","1");
        platformInfoQueryWrapper.eq("env",pushProperties.getEnv());
        List<PlatformInfo> platformInfos = platformInfoService.list(platformInfoQueryWrapper);
        for (PlatformInfo platformInfo :platformInfos) {
            String timeStamp = String.valueOf(new Date().getTime());
            String seq = SeqUtils.getUniqueInstance().getSeq();
            String signParams = PileSignUtils.signParams(JSONObject.toJSONString(chargeStatus), platformInfo.getDataSecret(), platformInfo.getDataSecretIv());
            String sign = HmacMd5Signature.hmacMd5(platformInfo.getJingliPlatformId()+signParams+timeStamp+seq,platformInfo.getSignSecret());
            CommonDataResult commonDataResult = new CommonDataResult();
            commonDataResult.setPlatformId(platformInfo.getJingliPlatformId());
            commonDataResult.setSeq(seq);
            commonDataResult.setTimeStamp(timeStamp);
            commonDataResult.setRet(0);
            commonDataResult.setMsg("");
            commonDataResult.setData(signParams);
            commonDataResult.setSig(sign);
            String s = thirdParty.notifyConnectorStatus(platformInfo.getPlatformId(), platformInfo.getApiUrl(), JSONObject.toJSONString(commonDataResult));
            logger.info("充电状态推送结果:{}",s);
        }

    }

    //推送设备接口状态信息
//    @KafkaListener(topics = "${topic.name.portStatus}")
    public void pushPortStatus(String content) throws Exception {
        logger.info("收到接口状态变更信息:{}",content);
        KafkaPortStatus kafkaPortStatus = JSONObject.parseObject(content, KafkaPortStatus.class);
        SupConnectorStatusInfo statusInfo = new SupConnectorStatusInfo();
        SupConnectorInfo supConnectorInfo = connectorInfoService.getById(kafkaPortStatus.getPortId());
        SupEquipmentInfo equipmentInfo = equipmentInfoService.getById(supConnectorInfo.getEquipmentId());
        SupStationInfo stationInfo = stationInfoService.getById(equipmentInfo.getStationId());
        statusInfo.setConnectorId(kafkaPortStatus.getPortId());
        statusInfo.setOperatorId(stationInfo.getOperatorId());
        statusInfo.setEquipmentClassification(1);
        statusInfo.setEquipmentOwnerId(stationInfo.getEquipmentOwnerId());
        statusInfo.setStationId(stationInfo.getStationId());
        statusInfo.setEquipmentId(equipmentInfo.getEquipmentid());
//        statusInfo.setBatteryStatus(Integer.valueOf(kafkaPortStatus.getPortStatus()));
        statusInfo.setBatteryPackId("");
        statusInfo.setStatus(Objects.equals(kafkaPortStatus.getPortStatus(), "1") ?3:1);
        statusInfo.setLastChangeTime(DateUtil.formatDateTime(new Date()));
        connectorStatusInfoService.saveOrUpdate(statusInfo);
        logger.info("准备推送接口状态变更信息:{}",JSONObject.toJSONString(statusInfo));
        QueryWrapper<PlatformInfo> platformInfoQueryWrapper = new QueryWrapper<>();
        platformInfoQueryWrapper.eq("push_enable","1");
        platformInfoQueryWrapper.eq("env",pushProperties.getEnv());
        List<PlatformInfo> platformInfos = platformInfoService.list(platformInfoQueryWrapper);
        for (PlatformInfo platformInfo :platformInfos) {
            String timeStamp = String.valueOf(new Date().getTime());
            String seq = SeqUtils.getUniqueInstance().getSeq();
            String signParams = PileSignUtils.signParams(JSONObject.toJSONString(statusInfo), platformInfo.getDataSecret(), platformInfo.getDataSecretIv());
            String sign = HmacMd5Signature.hmacMd5(platformInfo.getJingliPlatformId()+signParams+timeStamp+seq,platformInfo.getSignSecret());
            CommonDataResult commonDataResult = new CommonDataResult();
            commonDataResult.setPlatformId(platformInfo.getJingliPlatformId());
            commonDataResult.setSeq(seq);
            commonDataResult.setTimeStamp(timeStamp);
            commonDataResult.setRet(0);
            commonDataResult.setMsg("");
            commonDataResult.setData(signParams);
            commonDataResult.setSig(sign);
            String s = thirdParty.notifyConnectorChangeStatus(platformInfo.getPlatformId(), platformInfo.getApiUrl(), JSONObject.toJSONString(commonDataResult));
            logger.info("推送充电设备接口状态信息:{}",s);
        }

    }

}
